package com.mdp.im.core.queue;

import com.alibaba.fastjson.JSONObject;
import com.mdp.im.core.entity.ImGroupMessage;
import com.mdp.im.core.entity.ImPrichatMessage;
import com.mdp.im.core.entity.ImSysMessage;
import com.mdp.im.core.service.ImGroupMessageService;
import com.mdp.im.core.service.ImPrichatMessageService;
import com.mdp.im.core.service.ImSysMessageService;
import com.mdp.im.core.utils.StoreQueueNames;
import com.mdp.im.core.scbscriber.api.StoreSubscriber;
import com.mdp.mq.queue.Pop;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 从队列获取需要入库的消息，存入数据库
 * 
 * @author Leo
 * @date 2018/3/29
 */
@Service
public class StoreMessageSubscriber implements StoreSubscriber {
	
	 private static final Logger logger = LoggerFactory.getLogger(StoreMessageSubscriber.class);

	ExecutorService e = Executors.newFixedThreadPool(50);

	Long timeOut= Long.valueOf(60000);

	@Autowired
	ImGroupMessageService groupMessageService;
	@Autowired
	ImSysMessageService sysMessageService;
	@Autowired
	ImPrichatMessageService prichatMessageService;

	@Autowired
	Pop pop;

	
	@Override
	public void onSysMessageFromDbQueue(String message) { 
		ImSysMessage messageDto=JSONObject.parseObject(message, ImSysMessage.class);
		sysMessageService.insert(messageDto);
	    
	}
	@Override
	public void onPrichatMessageFromDbQueue(String message) {
		ImPrichatMessage messageDto=JSONObject.parseObject(message, ImPrichatMessage.class);
		prichatMessageService.insert(messageDto);
		
	}
	@Override
	public void onGroupMessageFromDbQueue(String message) { 
        ImGroupMessage messageDto=JSONObject.parseObject(message, ImGroupMessage.class);
        groupMessageService.insert(messageDto);
		
	}
	@PostConstruct
	public void startGroupMessageSubscriberFromDbQueue() {

		e.execute(new Runnable() {
			public void run() {
				while(true) {
					try {
						String  message=(String) pop.rightPop(StoreQueueNames.im_group_message_queue_name, timeOut);
						if(StringUtils.isEmpty(message)){
							continue;
						}
						onGroupMessageFromDbQueue(message);
					} catch (Exception e) {

					}
				}

			}
		});
	}
	/**
	 * 自动订阅
	 */
	@PostConstruct
	public void startPrichatMessageSubscriberFromDbQueue() {
		e.execute(new Runnable() {
			public void run() {
				while(true) {
					try {
						String  message=(String) pop.rightPop(StoreQueueNames.im_prichat_message_queue_name, timeOut);
						if(StringUtils.isEmpty(message)){
							continue;
						}
						onPrichatMessageFromDbQueue(message);
					} catch (Exception e) {

					}
				}

			}
		});
	}
	/**
	 * 自动订阅
	 */
	@PostConstruct
	public void startSysMessageSubscriberFromDbQueue() {
		e.execute(new Runnable() {
			public void run() {
				while(true) {
					try {
						String  message=(String)pop.rightPop(StoreQueueNames.im_sys_message_queue_name, timeOut);
						if(StringUtils.isEmpty(message)){
							continue;
						}
						onSysMessageFromDbQueue(message);
					} catch (Exception e) {

					}
				}

			}
		});
	}
}
